package com.sunlands.deskmate.channel;

import com.sunlands.deskmate.client.UserCenterService;
import com.sunlands.deskmate.constant.YN;
import com.sunlands.deskmate.exception.SendExceptionUtil;
import com.sunlands.deskmate.util.SpringUtils;
import com.sunlands.deskmate.utils.IdGenerator;
import com.sunlands.deskmate.vo.OnlineStateVO;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMapCache;
import org.redisson.api.RedissonClient;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import static com.sunlands.deskmate.config.RedisConfig.PREFIX;

/**
 * @author anjunling@sunlands.com
 */
@Slf4j
public class ChannelManager {

    /**
     * 多节点的情况下, 只记录当前节点下登录的长连接信息
     * sessionKey -> Channel
     * channel本地仓库
     */
    private static final Map<String, Channel> localChannelStore = new ConcurrentHashMap<>();

    /**
     * redis channel key 前缀
     */
    private static final String REDIS_CHANNEL_SET = PREFIX + ":channel:set";
    /**
     * redis channel key过期时间，设置0为永不过期 单位：分钟
     */
    public static final long REDIS_CHANNEL_TIME_OUT = 0;
    /**
     * redis channel活跃时长，超过时长没有任何操作，自动删除，单位：分钟
     */
    public static final long REDIS_CHANNEL_MAX_IDLE_TIME = 5;
    /**
     * 心跳时长
     */
    public static final long HEART_INTERVAL = 30;//单位：秒

    private volatile static RedissonClient redissonClient;

    private volatile static SendExceptionUtil sendExceptionUtil;

    private volatile static UserCenterService userCenterService;

    private volatile static ThreadPoolTaskExecutor threadPoolTaskExecutor;

    public static final AttributeKey<Integer> TOP_KEY = AttributeKey.newInstance("TOP_KEY");
    public static final AttributeKey<Long> HEART_TIME = AttributeKey.newInstance("HEART_TIME");
    public static final AttributeKey<ChannelInfo> CHANNEL_INFO = AttributeKey.valueOf("info");
    private static final String UPDATE_USER_ONLINE_STATE_LOCK = "desk-mate-socket:user:online:update:lock";
    /**
     * channel关闭时间，防止程序主动关闭时，异步通知创建时间延迟
     */
    public static final AttributeKey<Long> CHANNEL_CREATE_TIME = AttributeKey.valueOf("channelCreateTime");

    private ChannelManager() {
    }

    public static Map<String, Channel> getAllChannel() {
        return localChannelStore;
    }

    /**
     * 新增通道
     *
     * @param channelInfo
     * @param channel
     */
    public static void add(ChannelInfo channelInfo, Channel channel) {
        String sessionKey = channelInfo.getSessionKey();

        Channel cachedChannel = localChannelStore.get(sessionKey);

        log.debug("release previous when created sessionKey={}, previousChannel={}, currentChannel={}", channelInfo.getSessionKey(), cachedChannel, channel);
        //关闭之前的链接
        if (Objects.nonNull(cachedChannel)) {
            release(cachedChannel);
            log.info("之前存在通道，关闭旧通道 cachedChannel = {}", cachedChannel);
        }

        //设置channel创建时间，用于防止登录、退出通知乱序导致的用户登录状态混乱
        long now = System.currentTimeMillis();
        channelInfo.setCreateTime(now);
        channelInfo.setWorkId(IdGenerator.getWorkerId());
        channel.attr(CHANNEL_CREATE_TIME).set(now);
        channel.attr(CHANNEL_INFO).set(channelInfo);
        channel.attr(HEART_TIME).set(now);

        //channel存入本地仓库
        Channel absent = localChannelStore.putIfAbsent(channelInfo.getSessionKey(), channel);
        if (Objects.nonNull(absent)) {
            log.info("存入本地仓库失败，直接关闭 channel = {}", channel);
            channel.close();
            return;
        }

        //channel存入远程仓库
        RMapCache<String, ChannelInfo> channelRedisSet = getRedisChannelStore();
        ChannelInfo exist = channelRedisSet.get(channelInfo.getSessionKey());
        if (Objects.isNull(exist)) {

            //设置key的活跃时间，超时自动删除key
            ChannelInfo info = channelRedisSet.putIfAbsent(channelInfo.getSessionKey(),
                    channelInfo, REDIS_CHANNEL_TIME_OUT,
                    TimeUnit.DAYS, REDIS_CHANNEL_MAX_IDLE_TIME,
                    TimeUnit.MINUTES);

            if (Objects.nonNull(info)) {
                channel.close();
                log.warn("putIfAbsent 异常情况，远程通道创建失败 exist = {} channelInfo = {}", exist, channelInfo);
            }
        } else if (exist.getCreateTime() < channelInfo.getCreateTime()) {

            boolean remove = channelRedisSet.remove(channelInfo.getSessionKey(), exist);
            ChannelInfo ifAbsent = channelRedisSet.putIfAbsent(channelInfo.getSessionKey(),
                    channelInfo,
                    REDIS_CHANNEL_TIME_OUT,
                    TimeUnit.DAYS,
                    REDIS_CHANNEL_MAX_IDLE_TIME,
                    TimeUnit.MINUTES);

            if (!(remove && Objects.isNull(ifAbsent))) {
                channel.close();
                log.warn("replace 异常情况，远程通道替换失败 exist = {} channelInfo = {}", exist, channelInfo);
            }
        } else {
            channel.close();
            log.warn("exist 异常情况，已建立通道晚于当前通道 exist = {} channelInfo = {}", exist, channelInfo);
        }
        log.info("通道操作-通道创建成功 channelInfo = {}", channelInfo);

        //设置用户服务中用户状态
        userOnline(channelInfo.getUserId(), channelInfo.getTopRun(), channelInfo.getCreateTime(), 0L);
        log.info("通道创建成功 channelInfo = {}", channelInfo);
    }

    public static RMapCache<String, ChannelInfo> getRedisChannelStore() {
        return getRedissonClient().getMapCache(REDIS_CHANNEL_SET);
    }

    private static RedissonClient getRedissonClient() {
        if (Objects.isNull(redissonClient)) {
            synchronized (ChannelManager.class) {
                if (Objects.isNull(redissonClient)) {
                    redissonClient = SpringUtils.getApplicationContext().getBean(RedissonClient.class);
                }
            }
        }
        return redissonClient;
    }

    private static SendExceptionUtil getSendExceptionUtil() {
        if (Objects.isNull(sendExceptionUtil)) {
            synchronized (ChannelManager.class) {
                if (Objects.isNull(sendExceptionUtil)) {
                    sendExceptionUtil = SpringUtils.getApplicationContext().getBean(SendExceptionUtil.class);
                }
            }
        }
        return sendExceptionUtil;
    }


    private static UserCenterService getUserCenterService() {
        if (Objects.isNull(userCenterService)) {
            synchronized (ChannelManager.class) {
                if (Objects.isNull(userCenterService)) {
                    userCenterService = SpringUtils.getApplicationContext().getBean(UserCenterService.class);
                }
            }
        }
        return userCenterService;
    }

    public static void userOnline(Long userId, Integer topRun, Long channelId, Long backRunTime) {
        //将用户状态修改为在线
        OnlineStateVO onlineState = OnlineStateVO.builder()
                .userId(userId)
                .online(YN.Y.code)
                .topRun(topRun)
                .channelId(channelId)
                .backRunTime(backRunTime)
                .build();
        getThreadPoolTaskExecutor().execute(() -> {
            RLock lock = getRedissonClient().getLock(UPDATE_USER_ONLINE_STATE_LOCK + userId);
            try {
                if (lock.tryLock(3, TimeUnit.SECONDS)) {
                    log.info("用户上线，通知user 修改用户状态 onlineState = {}", onlineState);
                    getUserCenterService().updateOnline(onlineState);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (Objects.nonNull(lock) && lock.isLocked() && lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }

        });

    }

    public static void userOffline(Long userId, Long channelId) {
        //将用户状态修改为离线
        OnlineStateVO onlineState = OnlineStateVO.builder()
                .userId(userId)
                .online(YN.N.code)
                .channelId(channelId)
                .build();
        getThreadPoolTaskExecutor().execute(() -> {
            RLock lock = getRedissonClient().getLock(UPDATE_USER_ONLINE_STATE_LOCK + userId);
            try {
                if (lock.tryLock(3, 2, TimeUnit.SECONDS)) {
                    log.info("用户下线，通知user 修改用户状态 onlineState = {}", onlineState);
                    getUserCenterService().updateOnline(onlineState);
                }
            } catch (InterruptedException e) {
                log.error("", e);
            } finally {
                if (Objects.nonNull(lock) && lock.isLocked() && lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        });

    }

    private static ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
        if (Objects.isNull(threadPoolTaskExecutor)) {
            synchronized (ChannelManager.class) {
                if (Objects.isNull(threadPoolTaskExecutor)) {
                    threadPoolTaskExecutor = (ThreadPoolTaskExecutor) SpringUtils.getApplicationContext().getBean("userStateExecutor");
                }
            }
        }
        return threadPoolTaskExecutor;
    }

    public static Set<String> getAllSessionKeys() {
        return localChannelStore.keySet();
    }

    public static Channel getChannel(String sessionKey) {

        if (!localChannelStore.containsKey(sessionKey)) {
            return null;
        }

        return localChannelStore.get(sessionKey);
    }

    /**
     * 释放通道 由WebSocketServerHandler.channelInactive()触发调用
     *
     * @param channel
     */
    public static void release(Channel channel) {
        ChannelInfo channelInfo = null;
        try {

            if (Objects.isNull(channel)) {
                return;
            }

            channelInfo = channel.attr(CHANNEL_INFO).get();
            if (Objects.isNull(channelInfo)) {
                return;
            }

            log.info("关闭通道 channel, channelInfo={}, channel={}, channel size ={}", channelInfo, channel, localChannelStore.size());
            //程序主动关闭channel 设置channel创建时间，防止用户被顶下去时，由于异步通知消息导致登录状态更新错误
            //删除本地channel仓库中的通道信息
            Channel existChannel = localChannelStore.get(channelInfo.getSessionKey());
            if (Objects.nonNull(existChannel) && getChannelCreateTime(existChannel).equals(channelInfo.getCreateTime())) {
                log.info("通道操作-释放通道 channel, channelInfo={}, channel={}, channel size ={}", channelInfo, channel, localChannelStore.size());
                localChannelStore.remove(channelInfo.getSessionKey(), channel);
            }

            //删除远程redis channel仓库中的通道信息
            RMapCache<String, ChannelInfo> redisChannelSet = getRedisChannelStore();
            ChannelInfo exist = redisChannelSet.get(channelInfo.getSessionKey());
            if (Objects.nonNull(exist) && exist.getCreateTime() == channelInfo.getCreateTime()) {
                boolean remove = redisChannelSet.remove(exist.getSessionKey(), exist);
                log.info("移除远程通道结果 remove = {} channelInfo = {}", remove, channelInfo);
            }

            userOffline(channelInfo.getUserId(), channelInfo.getCreateTime());
        } finally {
            if (Objects.nonNull(channel) && channel.isActive()) {
                log.info("netty通道未关闭，再次关闭 channelInfo = {} channel {} is active, close", channelInfo, channel);
                channel.close();
            }
        }


    }

    public static ChannelInfo getChannelInfo(Channel channel) {
        if (channel != null) {
            return channel.attr(CHANNEL_INFO).get();
        }
        return null;
    }

    public static Long getChannelCreateTime(Channel channel) {
        if (channel != null) {
            return channel.attr(CHANNEL_CREATE_TIME).get();
        }
        return null;
    }

    public static Long getHeartTime(Channel channel) {
        if (channel != null) {
            return channel.attr(HEART_TIME).get();
        }
        return null;
    }

    public static Integer getTopKey(Channel channel) {
        if (channel != null) {
            return channel.attr(TOP_KEY).get();
        }
        return null;
    }


}
